package com.suxin.thread.schedulev2;

import com.suxin.thread.schedule.domain.TaskWithBlobs;
import com.suxin.thread.schedule.job.JobService;
import com.suxin.thread.schedule.task.Task;
import com.suxin.thread.schedule.task.TaskService;
import com.suxin.thread.schedulev2.dto.Response;
import com.suxin.thread.schedulev2.error.ExecutorErrorHandler;
import com.suxin.thread.schedulev2.error.InterruptedExecutorErrorHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author Tang
 * @classname JobExecutor
 * @description [ 任务流执行器 ]
 * @date 2021/6/10 13:12
 */
@Slf4j
public class JobExecutor {

    @Resource
    private JobService jobService;

    @Resource
    private TaskService taskService;

    private ExecutorService jobExecutorService = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors() + 1,
            Runtime.getRuntime().availableProcessors() + 2,500L,
            TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(30),new CustomizableThreadFactory("thread-executor-"));

    public Response execute(int jobId) {
        jobExecutorService.execute(() -> { this.execute0(jobId); });
        return new Response();
    }

    public void execute0(int jobId) {
        ThreadPoolExecutor taskExecutorThreadPool = JobExecutorContext.getCurrentTaskExecutorThreadPool(jobId);
        if (taskExecutorThreadPool != null) {
            log.warn("job 上次执行尚未完成，等待下次执行，jobId:{}",jobId);
            JobExecutorContext.sleepIfIdle(taskExecutorThreadPool);
            // 重新入队
            this.execute(jobId);
            return;
        }
        log.info("job 执行开始,jobId:{}",jobId);
        int logId = refreshAndLogStart(jobId);

        List<TaskWithBlobs> tasks = taskService.findAllByJobId(jobId);
        // 创建线程池
        taskExecutorThreadPool = buildTaskThreadPoolExecutor(jobId, tasks);
        JobExecutorContext.registerTaskExecutorThreadPool(jobId,taskExecutorThreadPool);
        CountDownLatch countDownLatch = new CountDownLatch(tasks.size());
        // 错误回执
        ExecutorErrorHandler executorErrorHandler = new InterruptedExecutorErrorHandler();
        for (TaskWithBlobs task : tasks) {
            setSateOfTaskToWaiting(task);
            taskExecutorThreadPool.execute(new TaskExecutorThread(logId, new Task(),executorErrorHandler,countDownLatch));
        }

        try {
            // 阻塞等待所有子任务执行完成
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    private void setSateOfTaskToWaiting(TaskWithBlobs task) {

    }

    private ThreadPoolExecutor buildTaskThreadPoolExecutor(int jobId, List<TaskWithBlobs> tasks) {
        return null;
    }

    private int refreshAndLogStart(int jobId) {
        return 0;
    }

}